package rocketmq

import (
	"context"
	"errors"
	"github.com/apache/rocketmq-clients/golang/v5"
	"shen/gotask/pkg/gotask"
	"shen/gotask/rocketmq/logs"
	"strconv"
	"time"
)

type RunConsumerCloseChan chan bool

type Result struct {
	NeedleRun   []ConsumerResult `json:"needle_run"`
	NeedleClose []string         `json:"needle_close"`
	Result      bool             `json:"result"`
}

type ConsumerResult struct {
	Class              string `json:"class"`
	Method             string `json:"method"`
	Topic              string `json:"topic"`
	Group              string `json:"group"`
	Num                int    `json:"num"`
	InvisibleDurations int    `json:"invisible_durations"`
	ModeType           string `json:"mode_type"`
	Mode               string `json:"mode"`
}

type ConsumerMessage struct {
	Body  any    `json:"body"`
	MsgId string `json:"msg_id"`
	Topic string `json:"topic"`
}

type ConsumerMessageResult struct {
	MsgId  string `json:"msg_id"`
	Result bool   `json:"result"`
}

type RunData struct {
	key int
}

type ConsumerParams struct {
	Name []string `json:"name"`
}

type Consumer struct {
	logs          *logs.Logs
	RunConsumerS  map[int]RunData
	isNeedleClose bool
}

func (c *Consumer) DoConsumer(consumerResult ConsumerResult, closeChan RunConsumerCloseChan, needleCloseConsumerChannel chan string) {
	defer func() {
		if err := recover(); err != nil {
			c.logs.ErrorFormat("消费者运行失败，错误信息：%v", err)
		}
	}()
	defer close(closeChan)
	c.logs = logs.NewLogs()
	c.RunConsumerS = make(map[int]RunData)
	var i int
	for {
		select {
		case <-closeChan:
			return
		default:
			if len(c.RunConsumerS) < consumerResult.Num {
				if c.isNeedleClose {
					needleCloseConsumerChannel <- consumerResult.Class + "::" + consumerResult.Method
					return
				} else {
					runData := RunData{
						key: i,
					}
					c.RunConsumerS[i] = runData
					i++
					c.logs.InfoFormat("正在启动消费者：%v::%v-%v", consumerResult.Class, consumerResult.Method, runData.key)
					go c.consumer(runData, consumerResult, closeChan)
				}
			} else {
				if c.isNeedleClose {
					needleCloseConsumerChannel <- consumerResult.Class + "::" + consumerResult.Method
					return
				}
				time.Sleep(3 * time.Second)
			}
		}
	}
}

func (c *Consumer) consumer(runData RunData, consumerResult ConsumerResult, closeChan RunConsumerCloseChan) {
	defer func() {
		if err := recover(); err != nil {
			c.logs.ErrorFormat("消费失败，%v", err)
		}
	}()
	defer delete(c.RunConsumerS, runData.key)
	name := consumerResult.Class + "::" + consumerResult.Method
	simpleConsumer, err := CreateConsumer(consumerResult.Topic, consumerResult.Group, consumerResult.Mode, c.getExpressionType(consumerResult.ModeType))
	if err != nil {
		return
	}
	defer simpleConsumer.GracefulStop()
	client, errClient := gotask.NewPhpClient()
	if errClient != nil {
		return
	}
	defer client.Close()

	for {
		select {
		case <-closeChan:
			c.logs.InfoFormat("正在关闭消费者：%v::%v-%v", consumerResult.Class, consumerResult.Method, runData.key)
			return
		default:
			mvs, errM := simpleConsumer.Receive(context.Background(), 2, time.Duration(consumerResult.InvisibleDurations)*time.Second)
			if c.isNeedleClose {
				return
			}
			if errM != nil {
				rpcErr := &golang.ErrRpcStatus{}
				if errors.As(errM, &rpcErr) && rpcErr.Code == 40401 {
					continue
				}
				c.logs.WarningFormat("Receive err: %v", errM)
				continue
			}

			for _, mv := range mvs {
				property := mv.GetProperties()
				isNeedleConsume := true
				if s, ok := property["ro-php-relay"]; ok {
					if relayTime, err := strconv.ParseInt(s, 10, 64); err == nil {
						if relayTime-(time.Now().Unix()) > 0 {
							if isPut := c.putMesToQueue(mv, relayTime); isPut {
								errAck := simpleConsumer.Ack(context.TODO(), mv)
								if errAck != nil {
									c.logs.Error("消息确认失败，消息id：" + mv.GetMessageId())
								} else {
									isNeedleConsume = false
								}
							}
						}
					} else {
						c.logs.Error("延时消息重新放入消息队列失败，消息id：" + mv.GetMessageId() + "错误信息：" + err.Error())
					}
				}

				if !isNeedleConsume {
					continue
				}

				message := ConsumerMessage{
					Body:  string(mv.GetBody()),
					MsgId: mv.GetMessageId(),
					Topic: mv.GetTopic(),
				}
				var res ConsumerMessageResult
				var retryNum = 0
			RETRY:
				errSend := client.Call(name, message, &res)
				if errSend != nil {
					if retryNum < 3 {
						client, errClient = gotask.NewPhpClient()
						retryNum++
						c.logs.Info("重试第" + strconv.Itoa(retryNum) + "次，消息id：" + mv.GetMessageId() + "name:" + name)
						goto RETRY
					}
					c.isNeedleClose = true
					return
				} else {
					if res.Result {
						errAck := simpleConsumer.Ack(context.TODO(), mv)
						if errAck != nil {
							c.logs.Error("消息确认失败，消息id：" + mv.GetMessageId())
						}
					}
				}
			}
		}
	}
}

func (c *Consumer) getExpressionType(modeType string) golang.FilterExpressionType {
	expressionType := golang.TAG
	switch modeType {
	case "tag":
		expressionType = golang.TAG
	case "sql":
		expressionType = golang.SQL92
	default:
		expressionType = golang.TAG
	}
	return expressionType
}

func (c *Consumer) putMesToQueue(mv *golang.MessageView, relayTime int64) bool {
	var msgSs ProducerMessage
	msgSs.Body = MsgBody{
		IsDelay:   true,
		DelayTime: int(relayTime),
		Message:   string(mv.GetBody()),
	}
	msgSs.Topic = mv.GetTopic()
	msgSs.Keys = mv.GetKeys()
	tag := mv.GetTag()
	if tag == nil {
		msgSs.Tag = ""
	} else {
		msgSs.Tag = *tag
	}

	producer := Producer{}
	_, err1 := producer.SendMsg(msgSs)
	if err1 != nil {
		c.logs.Error("延时消息重新放入消息队列失败，消息id：" + mv.GetMessageId() + "错误信息：" + err1.Error())
		return false
	} else {
		return true
	}
}
